package com.burt.flink.util;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import com.alibaba.fastjson.JSON;
import com.burt.flink.model.Student;

import lombok.extern.slf4j.Slf4j;

/**
 * <p>
 * Kafka工具类,往kafka中写数据,可以使用这个main函数进行测试
 * </p>
 *
 * @author Burt
 * @since 2019/9/8 0008 12:20
 */
@Slf4j
public class KafkaUtils {
    public static final String BROKER_LIST = "c-server-4:9092";
    /**
     * kafka topic需要和flink程序用同一个topic
     */
    public static final String TOPIC       = "student";

    public static void writeToKafka() throws InterruptedException {
        Properties prop = new Properties();
        prop.put("bootstrap.servers", BROKER_LIST);
        prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(prop)) {
            for (int i = 1; i <= 100; i++) {
                Student student = Student.builder().id(i).name("burt" + i).password("password" + i).age(18 + i).build();
                String data = JSON.toJSONString(student);
                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, null, null, data);
                producer.send(record);
                System.out.println("发送数据: " + data);
                //发送一条数据 sleep 10s，相当于 1 分钟 6 条
                Thread.sleep(10000);
            }
            producer.flush();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        writeToKafka();
    }
}
